package com.win.service.impl;

import com.win.common.HdfsCommon;
import com.win.pojo.vo.HdfsVo;
import com.win.service.HdfsService;
import com.win.utils.Result;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.stereotype.Service;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;

import static com.win.common.HdfsCommon.hdfsPath;

@Service
public class HdfsServiceImpl implements HdfsService {

    public Result getFile() {
        //获取hdfs配置对象
        Configuration configuration = new Configuration();

        List<HdfsVo> hdfsVoList = null;
        try {
            FileSystem fs = FileSystem.get(new URI(HdfsCommon.hdfsUri), configuration);
            List<String> fileNameList = getFileName(fs, hdfsPath);

            hdfsVoList = getFileContent(configuration,fileNameList, hdfsPath);
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (URISyntaxException e) {
            throw new RuntimeException(e);
        }

        return Result.ok(hdfsVoList);
    }



    public List<String> getFileName(FileSystem fs,String path){
        List<String> list = new ArrayList<>();

        // 获取文件系统实例
        try (fs) {
            // 获取指定路径下的所有文件和目录状态
            FileStatus[] status = fs.listStatus(new Path(path));

            // 遍历并打印文件/目录名
            for (FileStatus fileStatus : status) {
                Path filePath = fileStatus.getPath();
                if (fileStatus.isDirectory()) {
                    //System.out.println("Directory: " + filePath.getName());
                } else {
                    //System.out.println("File: " + filePath.getName());
                    list.add(filePath.getName());
                }
                // 如果需要递归列出子目录中的文件，可以添加递归逻辑
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        //System.out.println(hdfsFile);
        return list;
    }

    private List<HdfsVo> getFileContent(Configuration configuration,List<String> fileNameList,String path) throws IOException, URISyntaxException {

        List<HdfsVo> hdfsVoList = new ArrayList<>();

        StringBuilder sb = new StringBuilder();

        FileSystem fs = FileSystem.get(new URI(HdfsCommon.hdfsUri), configuration);

        for (String fileName : fileNameList) {
            Path filePath = new Path(path + "/" + fileName);
            FSDataInputStream inputStream = fs.open(filePath);
            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));

            String line;
            for (int i = 0; (line = reader.readLine()) != null && i < 5; i++) {
                //System.out.println(line);
                sb.append(line);
            }
            //System.out.println(sb.toString());
            HdfsVo hdfsVo = HdfsVo.builder()
                    .fileName(fileName)
                    .fileContent(sb.toString())
                    .build();
            hdfsVoList.add(hdfsVo);
        }
        return hdfsVoList;
    }
}
